package com.ht.api.flink;

//import java.io.Serializable;
//import java.util.ArrayList;
//import java.util.Base64;
//import java.util.List;
//import java.util.Properties;
//
//import org.apache.flink.api.common.eventtime.WatermarkStrategy;
//import org.apache.flink.api.common.functions.MapFunction;
//import org.apache.flink.streaming.api.CheckpointingMode;
//import org.apache.flink.streaming.api.environment.CheckpointConfig;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.http.client.config.RequestConfig;
//import org.apache.http.client.methods.CloseableHttpResponse;
//import org.apache.http.client.methods.HttpPut;
//import org.apache.http.entity.StringEntity;
//import org.apache.http.impl.client.CloseableHttpClient;
//import org.apache.http.impl.client.HttpClients;
//import org.apache.http.impl.client.LaxRedirectStrategy;
//import org.apache.http.util.EntityUtils;
//
//import com.ververica.cdc.connectors.mysql.source.MySqlSource;
//import com.ververica.cdc.connectors.mysql.table.StartupOptions;
//import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
//
//import cn.hutool.json.JSONObject;
//import cn.hutool.json.JSONUtil;
//import lombok.Data;
//import lombok.extern.slf4j.Slf4j;

//@Slf4j
public class MysqlToSelectDB {
//
//    @Data
//    public static class Config implements Serializable{
//        private static final long serialVersionUID = 1L;
//        // MySQL配置
//        private String hostname = "127.0.0.1";
//        private String username = "root";
//        private String password = "aixi_2028";
//        private int port = 3306;
//        private String database = "test";
//        private String table = "users";
//
//        // SelectDB配置
//        private String feHost = "selectdb-cn-rp64247ug01-public.selectdbfe.rds.aliyuncs.com";
//        private int fePort = 8080;  
//        private String dbName = "test";
//        private String tableName = "users";
//        private String selectDBUsername = "admin";
//        private String selectDBPassword = "xx&*(%^Ht20241234yuiexx";
//    }
//    
//    // 定义静态内部类
//    private static class CDCMapFunction implements MapFunction<String, String>, Serializable {
//        private static final long serialVersionUID = 1L;
//        private final Config config;
//        private final List<String> batch = new ArrayList<>();
//        private static final int BATCH_SIZE = 10000; // 批量大小
//        private long lastFlushTime = System.currentTimeMillis();
//        private static final long FLUSH_INTERVAL = 100000; // 5秒
//
//        public CDCMapFunction(Config config) {
//            this.config = config;
//        }
//
//        private void sendBatchToSelectDB(List<String> records) {
//            if (records.isEmpty()) {
//                return;
//            }
//
//            String streamLoadUrl = String.format("http://%s:%d/api/%s/%s/_stream_load",
//                    config.getFeHost(), config.getFePort(), config.getDbName(), config.getTableName());
//
//            RequestConfig requestConfig = RequestConfig.custom()
//                    .setExpectContinueEnabled(true)
//                    .setConnectTimeout(100000)
//                    .setSocketTimeout(100000)
//                    .setRedirectsEnabled(true)
//                    .setMaxRedirects(3)
//                    .setCircularRedirectsAllowed(true)
//                    .build();
//
//            try (CloseableHttpClient httpClient = HttpClients.custom()
//                    .setDefaultRequestConfig(requestConfig)
//                    .setRedirectStrategy(new LaxRedirectStrategy())
//                    .build()) {
//
//                HttpPut httpPut = new HttpPut(streamLoadUrl);
//
//                // 设置认证头
//                String auth = config.getSelectDBUsername() + ":" + config.getSelectDBPassword();
//                String encodedAuth = Base64.getEncoder().encodeToString(auth.getBytes());
//                httpPut.setHeader("Authorization", "Basic " + encodedAuth);
//
//                // 修改头信息
//                httpPut.setHeader("format", "json");
//                httpPut.setHeader("strip_outer_array", "true");
//                httpPut.setHeader("columns", "id,name,sex,age");  // 确保列名与 StarRocks 表结构完全匹配
//                httpPut.setHeader("Content-Type", "application/json");
//                httpPut.setHeader("Expect", "100-continue");
//                
//                // 修改数据质量相关参数
//                httpPut.setHeader("max_filter_ratio", "1.0");     // 允许所有行被过滤
//                httpPut.setHeader("timeout", "600");              // 设置超时时间为600秒
//                httpPut.setHeader("merge_type", "APPEND");        // 使用APPEND模式
//                httpPut.setHeader("jsonpaths", "[\"$.id\",\"$.name\",\"$.sex\",\"$.age\"]");  // 指定JSON路径
//                httpPut.setHeader("json_root", "");               // 不使用JSON根节点
//
//                // 构建请求体，确保JSON格式正确
//                StringBuilder jsonData = new StringBuilder("[");
//                for (int i = 0; i < records.size(); i++) {
//                    if (i > 0) {
//                        jsonData.append(",");
//                    }
//                    // 确保每条记录是一个有效的JSON对象
//                    String record = records.get(i);
//                    if (!record.startsWith("{")) {
//                        record = "{" + record + "}";
//                    }
//                    jsonData.append(record);
//                }
//                jsonData.append("]");
//
//                StringEntity entity = new StringEntity(jsonData.toString(), "UTF-8");
//                entity.setContentType("application/json");
//                httpPut.setEntity(entity);
//
//                // 执行请求
//                try (CloseableHttpResponse response = httpClient.execute(httpPut)) {
//                    int statusCode = response.getStatusLine().getStatusCode();
//                    String responseBody = EntityUtils.toString(response.getEntity());
//                    log.info("Batch Stream Load response: {} - {}", statusCode, responseBody);
//
//                    if (statusCode == 307) {
//                        String location = response.getFirstHeader("Location").getValue();
//                        HttpPut redirectPut = new HttpPut(location);
//                        redirectPut.setHeaders(httpPut.getAllHeaders());
//                        redirectPut.setEntity(entity);
//
//                        try (CloseableHttpResponse redirectResponse = httpClient.execute(redirectPut)) {
//                            String redirectBody = EntityUtils.toString(redirectResponse.getEntity());
//                            log.info("Redirect response: {} - {}", 
//                                    redirectResponse.getStatusLine().getStatusCode(),
//                                    redirectBody);
//                        }
//                    }
//                }
//            } catch (Exception e) {
//                log.error("批量处理失败", e);
//            }
//        }
//
//        @Override
//        public String map(String value) throws Exception {
//            try {
//                JSONObject json = JSONUtil.parseObj(value);
//                JSONObject after = json.getJSONObject("after");
//                String op = json.getStr("op");
//
//                if ("c".equals(op) || "u".equals(op)) {
//                    // 构建符合 StarRocks 表结构的 JSON 对象
//                    JSONObject data = new JSONObject();
//                    data.set("id", after.getInt("id"));
//                    data.set("name", after.getStr("name"));
//                    data.set("sex", after.getInt("sex"));
//                    data.set("age", after.getInt("age"));
//                    
//                    batch.add(data.toString());
//                    
//                    if (batch.size() >= BATCH_SIZE || 
//                        System.currentTimeMillis() - lastFlushTime > FLUSH_INTERVAL) {
//                        sendBatchToSelectDB(new ArrayList<>(batch));
//                        batch.clear();
//                        lastFlushTime = System.currentTimeMillis();
//                    }
//                }
//                return value;
//            } catch (Exception e) {
//                log.error("处理消息失败: {}", value, e);
//                return null;
//            }
//        }
//    }
//
//    public static void main(String[] args) throws Exception {
//        Config config = new Config();
//        
//        // 1. 设置环境
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(3);  // 根据实际情况调整
//        env.getConfig().setAutoWatermarkInterval(1000);
//        env.getConfig().setLatencyTrackingInterval(5000);
//        
//        // 设置检查点
//        env.enableCheckpointing(3000);
//        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
//        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        checkpointConfig.setMinPauseBetweenCheckpoints(500);
//        checkpointConfig.setCheckpointTimeout(60000);
//        checkpointConfig.setMaxConcurrentCheckpoints(1);
//        
//        // 2. MySQL Source
//        Properties jdbcProperties = new Properties();
//        jdbcProperties.setProperty("connectTimeout", "30000");
//        jdbcProperties.setProperty("socketTimeout", "30000");
//        jdbcProperties.setProperty("useSSL", "false");
//
//        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
//            .hostname(config.getHostname())
//            .port(config.getPort())
//            .databaseList(config.getDatabase())
//            .tableList(config.getDatabase() + "." + config.getTable())
//            .username(config.getUsername())
//            .password(config.getPassword())
//            .deserializer(new JsonDebeziumDeserializationSchema())
//            .startupOptions(StartupOptions.initial())
//            .serverId("5400-5440")
//            .jdbcProperties(jdbcProperties)
//            .build();
//
//        // 3. 数据处理并写入SelectDB
//        // 使用静态内部类替代匿名内部类
//        env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL CDC")
//           .map(new CDCMapFunction(config));
//           
//        // 4. 执行
//        env.execute("MySQL CDC to SelectDB Job");
//        
//        // 5. 插入测试数据
//    }
}
